package com.kongxiang.storm.bolt;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import com.kongxiang.storm.constant.Descid;

import java.io.Serializable;
import java.util.Map;

/**
 *  派发数据
 */
public class DistributeBolt implements IRichBolt,Serializable {

    private static final long serialVersionUID = 1L;

    OutputCollector collector;

    // 具体执行 分发数据
    @Override
    public void execute(Tuple input) {

         // 获取数据
        String message = input.getString(0);
        // 请求
        if(message!= null && message.indexOf("request")>-1 ){
            collector.emit(Descid.REQUEST_STREAM_ID.name(),new Values(message));
        }
        //异常
        else if(message !=null && message.indexOf("exception")>-1){
            collector.emit(Descid.EXCEPTION_STREAM_ID.name(),new Values(message));
        }
        else{
            //其他情况
            collector.emit(Descid.OTHER_STREAM_ID.name(),new Values(message));
        }


    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // 请求
        declarer.declareStream(Descid.REQUEST_STREAM_ID.name(), new Fields("R_MSG"));
        // 异常类
        declarer.declareStream(Descid.EXCEPTION_STREAM_ID.name(), new Fields("E_MSG"));
        // 其他
        declarer.declareStream(Descid.OTHER_STREAM_ID.name(), new Fields("O_MSG"));
    }

    @Override
    @SuppressWarnings("rawtypes")
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void cleanup() {
      // 回收资源
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
